package com.geek.gkpush.biz.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.geek.gkpush.biz.event.order.FoodOrderEventMsgHandler;
import com.geek.gkpush.dal.bean.EventMsgDO;
import com.geek.gkpush.dal.mapper.EventMsgMapper;
import com.geek.gkpush.dal.model.EventMsgScanModel;
import com.geek.gkpush.framework.event.EventMsgPersistFacade;
import com.geek.gkpush.framework.event.enums.PushStatusEnum;
import com.geek.gkpush.framework.event.model.EventMsgUpdateModel;
import com.geek.gkpush.framework.job.AbstractLockSimpleJob;
import com.geek.gkpush.framework.retry.RetryMsgDispatch;
import com.geek.gkpush.framework.retry.model.RetryInsertModel;
import com.github.pagehelper.PageHelper;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <h1>事件消息调度任务。</h1><br/>
 *
 * <h1><br/><a href="https://u.geekbang.org/subject/arch3/100614201">Java 业务架构实战营：https://u.geekbang.org/subject/arch3/100614201</a></h1><br/><h1><a href="https://gitee.com/ylimhhmily/BizArchPracticeCamp">Java 业务架构实战营代码地址：https://gitee.com/ylimhhmily/BizArchPracticeCamp</a></h1><br/><h1><a href="https://u.geekbang.org/subject/java4th/1001148?source=app_share">高级 Java 工程体系课 2023 版：https://u.geekbang.org/subject/java4th/1001148?source=app_share</a></h1><br/><h1><a href="https://time.geekbang.org/column/intro/100312101?tab=catalog">Dubbo 源码剖析与实战：https://time.geekbang.org/column/intro/100312101?tab=catalog</a></h1>
 *
 * @author hmilyylimh
 *         ^_^
 * @version 0.0.1
 *         ^_^
 * @date 2023/10/2
 *
 */
@Slf4j
@Component
public class EventMsgJob extends AbstractLockSimpleJob {

    @Autowired
    private EventMsgMapper eventMsgMapper;
    @Autowired
    private RetryMsgDispatch retryMsgDispatch;
    @Autowired
    private EventMsgPersistFacade eventMsgPersistFacade;

    @Override
    public void doExecute(ShardingContext ctx) {
        // 构建检索数据模型
        EventMsgScanModel model = buildEventMsgScanModel(ctx);

        while (true) {
            // 1. 每次只查第一页数据
            PageHelper.startPage(1, model.getPageSize(), false);
            List<EventMsgDO> dbEventMsgList = queryShardingListByRangeCreateTime(model);
            if (CollectionUtils.isEmpty(dbEventMsgList)) {
                break;
            }

            // 2. 查出来的，如果小于 pageSize 个数的话，那就结束流程
            int size = dbEventMsgList.size();
            if (size < model.getPageSize()) {
                processEventMsgList(dbEventMsgList);
                break;
            }

            // 3. 刚好满一页的话，那就处理数据
            processEventMsgList(dbEventMsgList);

            // 4. 最后还不忘修改一下 endTimeMillis 的值，确保检索的范围是缩小的
            Long lastRecordCreateTime = dbEventMsgList.get(size - 1).getCreateTime();
            model.setEndTimeMillis(lastRecordCreateTime);
        }
    }

    private void processEventMsgList(List<EventMsgDO> dbEventMsgList) {
        Map<String, List<EventMsgDO>> bizTypeGroup = dbEventMsgList.stream().collect(Collectors.groupingBy(EventMsgDO::getBizType));
        for (Map.Entry<String, List<EventMsgDO>> entry : bizTypeGroup.entrySet()) {
            String bizType = entry.getKey();
            List<EventMsgDO> bizTypeOfEventMsgList = entry.getValue();

            // TODO 尝试调用批量重新触发远程推送

            // TODO 成功则修改状态
            update2Succ(bizTypeOfEventMsgList);

            // TODO 否则失败的话，则入库重试消息表
            transfer2RetryMsg(bizTypeOfEventMsgList);
        }
    }

    private void update2Succ(List<EventMsgDO> dbEventMsgList) {
        List<String> idList = dbEventMsgList.stream().map(EventMsgDO::getId).collect(Collectors.toList());
        eventMsgMapper.updateList2Succ(idList, PushStatusEnum.FAIL.getCode(), PushStatusEnum.SUCC.getMsg());
    }

    private void transfer2RetryMsg(List<EventMsgDO> dbEventMsgList) {
        for (EventMsgDO dbEventMsg : dbEventMsgList) {
            try {
                // 转换为重试消息所需要的对象模型
                RetryInsertModel insertModel = buildRetryInsertModel(dbEventMsg);

                // 入库重试消息表
                retryMsgDispatch.insertRetryMsg(insertModel);

                // 修改为 RETRY 状态
                update2Retry(dbEventMsg);
            } catch (Exception e) {
                log.error("dbEventMsg change to RETRY error.", e);
            }
        }
    }

    private EventMsgScanModel buildEventMsgScanModel(ShardingContext ctx) {
        EventMsgScanModel model = new EventMsgScanModel();

        model.setStatusList(Lists.newArrayList(PushStatusEnum.FAIL.getCode(), PushStatusEnum.INIT.getCode()));

        model.setShardingTotalCount(ctx.getShardingTotalCount());
        model.setShardingItemNo(ctx.getShardingItem());

        Date currDate = new Date();
        model.setBeginTimeMillis(DateUtils.addMinutes(currDate, -60).getTime());
        model.setEndTimeMillis(DateUtils.addMinutes(currDate, -1).getTime());

        model.setPageSize(1000);

        return model;
    }

    private RetryInsertModel buildRetryInsertModel(EventMsgDO dbEventMsg) {
        return new RetryInsertModel(
                dbEventMsg.getBizId(),
                dbEventMsg.getId(),
                FoodOrderEventMsgHandler.class,
                5);
    }

    private int update2Retry(EventMsgDO dbEventMsg) {
        EventMsgUpdateModel model  = new EventMsgUpdateModel();

        model.setId(dbEventMsg.getId());
        model.setBizId(dbEventMsg.getBizId());
        model.setBizType(dbEventMsg.getBizType());

        model.setPushStatus(PushStatusEnum.RETRY.getCode());
        model.setSrcPushStatus(dbEventMsg.getPushStatus());
        model.setPushResult(PushStatusEnum.RETRY.getMsg());

        // 修改为 RETRY 成功状态
        return eventMsgPersistFacade.updatePushStatusByCondition(model);
    }

    private List<EventMsgDO> queryShardingListByRangeCreateTime(EventMsgScanModel model) {
        return eventMsgMapper.queryShardingListByRangeCreateTime(model);
    }
}